package org.example.mq.consume;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.example.idempotent.mq.NoMQDuplicateConsume;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

@Component
@RocketMQMessageListener(
        topic = "TEST_IDEMPOTENT_TOPIC",
        consumerGroup = "TEST_IDEMPOTENT_GROUP",
        selectorExpression = "*"
)
@Slf4j
public class RocketMQConsumer implements RocketMQListener<MessageExt> {

    @SneakyThrows
    @NoMQDuplicateConsume(
            uniqueKey = "#messageExt.properties['messageId']",
            keyTimeout = 120
    )
    @Override
    public void onMessage(MessageExt messageExt) {
        // 获取消息唯一标识：优先使用自定义messageId，其次使用RocketMQ自带msgId
        String messageId = messageExt.getUserProperty("messageId");
        if (messageId == null) {
            messageId = messageExt.getMsgId(); // RocketMQ生成的唯一消息ID
        }

        byte[] body = messageExt.getBody();
        if (body == null || body.length == 0) {
            log.info("消息体为空，messageId: {}" , messageId);
            return;
        }

        String str = new String(body, "UTF-8");
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy年MM月dd日 HH时mm分ss秒");

        log.info("messageId: {}, 接收到消费消息：{}，当前接收时间：{}", messageId, str, dateFormat.format(new Date()));
    }

}
